package queue;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.nio.channels.WritableByteChannel;
import java.util.Arrays;
import java.util.Set;
import java.util.concurrent.Executor;

import com.google.protobuf.CodedOutputStream;
import com.google.protobuf.Message;

public class NonBlockingClient implements QueueFactory {

	private final Selector selector;
	private final SocketChannel socketChannel;
	
	private boolean running;
	
	//@Property
	private String hostname;
	public NonBlockingClient setHostname(String hostname)
		{ this.hostname = hostname; return this;}
	
	//@Property
	private int port;
	public NonBlockingClient setPort(int port)
		{ this.port = port; return this; }

	private Executor executor;
	public NonBlockingClient setExecutor(Executor executor)
		{ this.executor = executor; return this; }
	
	public NonBlockingClient() {
		try {
			selector = Selector.open();
			socketChannel = SocketChannel.open();
			socketChannel.configureBlocking(false);
		} catch (IOException e) {
			throw new RuntimeException(e);
		}
	}
	
	public void stop() {
		running = true;
	}
	
	public void start() {
		running = true;
		connect();
		executor.execute(new Runnable() {
			public void run() {
				while(running) {
					try {
						selector.selectNow();
					} catch (IOException e) {
						throw new RuntimeException(e);
					}
					Set<SelectionKey> keys = selector.selectedKeys();
					for(SelectionKey selectedKey : keys) {
						final SelectionKey key = selectedKey;
						if(key.isValid() && key.isWritable()) {
							keys.remove(selectedKey);
							executor.execute(new Runnable() {
								public void run() {
									WritableByteChannel channel = (WritableByteChannel) key.channel();
									Message message = (Message) key.attachment();
									try {
										byte[] buf = new byte[message.getSerializedSize() + CodedOutputStream.computeRawVarint32Size(message.getSerializedSize())];
										CodedOutputStream cos = CodedOutputStream.newInstance(buf);
										cos.writeRawVarint32(message.getSerializedSize());
										ByteBuffer b = ByteBuffer.wrap(buf);
										b.put(message.toByteArray());
										cos.flush();
										b.flip();
										System.out.println(String.format("Client: %s", Arrays.toString(b.array())));
										channel.write(b);
										key.cancel();
									} catch (IOException e) {
										throw new RuntimeException(e);
									}
								}});								
						}
					}
				}
			}});
	}
	
	private void connect() {
		try {
			socketChannel.register(selector, SelectionKey.OP_CONNECT);
			if(socketChannel.connect(new InetSocketAddress(hostname, port))) {
				return;
			}
		} catch (ClosedChannelException e) {
			throw new RuntimeException(e);
		} catch (IOException e) {
			throw new RuntimeException(e);		
		}
		executor.execute(new Runnable() {
			private boolean connected;
			public void run() {
				while(running && !connected) {
					try {
						selector.selectNow();
					} catch (IOException e) {
						throw new RuntimeException(e);	
					}
					Set<SelectionKey> keys = selector.selectedKeys();
					for(SelectionKey key : keys) {
						if(key.isValid() && key.isConnectable()) {
							keys.remove(key);
							SocketChannel channel = (SocketChannel) key.channel();
							try {
								connected = channel.finishConnect();
								System.out.println("Channel succesfully connected!");
							} catch (IOException e) {
								throw new RuntimeException(e);	
							}
							if(connected) {
								key.cancel();
							}
						}
					}
				}
			}});
	}
	
	public Queue getQueue() {
		return new Queue() {
			public void put(Message message) {
				try {
					socketChannel.register(selector, SelectionKey.OP_WRITE, message);
				} catch (ClosedChannelException e) {
					throw new RuntimeException(e);
				}
			}};
	}
	
}
